Java 中的并发工具类

 发表于 2018-05-20 |  分类于学习笔记

 字数统计 3,983 |  阅读时长 17

java.util.concurrent 下提供了一些辅助类来帮助我们在并发编程的设计。

学习了 AQS 后再了解这些工具类，就非常简单了。

jdk 1.8

等待多线程完成的CountDownLatch

在 concurrent 包下面提供了 CountDownLatch 类，它提供了计数器的功能，能够实现让一个线程等待其他线程执行完毕才能进入运行状态。

源码分析

首先看下最关键的地方它的自定义同步器的实现，非常简单：

  private static final class Sync extends AbstractQueuedSynchronizer {
      private static final long serialVersionUID = 4982264981922014374L;
// 1. 初始化 state 资源变量
      Sync(int count) {
          setState(count);
      }
   
      int getCount() {
          return getState();
      }
// 尝试获取贡献模式下的资源，
      // 定义返回值小于 0 的时候获取资源失败
      protected int tryAcquireShared(int acquires) {
          return (getState() == 0) ? 1 : -1;
      }
   
      protected boolean tryReleaseShared(int releases) {
          // Decrement count; signal when transition to zero
          // 自旋。
          for (;;) {
              int c = getState();
              if (c == 0)
                  return false;
              int nextc = c - 1; // 每次释放资源，硬编码减一个资源
              if (compareAndSetState(c, nextc))
                  return nextc == 0; // 知道为 0 的时候才释放成功，也就是所有线程必须都执行释放操作说明才释放成功。
           
          }
      }
  }

在这里查看构造器的源码得知，CountDownLatch 内部使用的是内部类Sync 继承了 AQS ，将我们传入进来的 count 数值当作 AQS state。感觉这个是不是和可重入锁实现是一样的，只不过开始指定了线程获取的锁的次数。

在上面我也发现了几个特点，第一次看这个代码其实还是不好理解，因为它相对前面的 AQS 和 TwinsLock 就是一个反着设计的代码：

首先获取资源的时候，线程全部都是先进入等待队列，而且在这一步骤中，不改变 state 资源的数量；
释放资源的时候，每次固定减少一个资源，直到资源为 0 的时候才表示释放资源成功，所以加入我们有 5 个资源，但是只有四个线程执行，如果只释放四次（总共执行 countDown 四次），就永远也释放不成功，await 一直在阻塞。
经过上面的分析，发现了 state 的资源数量每次进行 countDown 都去减少一个，没有方法去增加数量，所以它是不可逆的，它的计数器是不可以重复使用的。

看下 await 的实现，发现它最终实现的是 doAcquireSharedInterruptibly ：

   // 仔细看这个代码，和前面的共享模式中的 doAcquireShared 方法基本一摸一样，只不过是当它遇到线程中断信号的时候，立刻抛出中断异常，仔细想想也是的，比如，自己在这里等别人吃饭，不想等了，也懒得管别人做什么了，剩下的吃饭的事情也没必要继续下去了。
private void doAcquireSharedInterruptibly(int arg)
       throws InterruptedException {
       final Node node = addWaiter(Node.SHARED);
       boolean failed = true;
       try {
           for (;;) {
               final Node p = node.predecessor();
               if (p == head) {
                   int r = tryAcquireShared(arg);
                   if (r >= 0) {
                       setHeadAndPropagate(node, r);
                       p.next = null; // help GC
                       failed = false;
                       return;
                   }
               }
               if (shouldParkAfterFailedAcquire(p, node) &&
                   parkAndCheckInterrupt())
                   throw new InterruptedException();
           }
       } finally {
           if (failed)
               cancelAcquire(node);
       }
   }
	// 需要注意的是它重写了尝试获取资源的方法，当资源全部消耗完，才能够让你去获取资源，现在才豁然开朗，await 阻塞的线程就是这么被唤醒的。
       protected int tryAcquireShared(int acquires) {
           return (getState() == 0) ? 1 : -1;
       }

使用场景

CountDownLatch允许一个或多个线程等待其他线程完成操作。

比如经典问题：

有Thread1、Thread2、Thread3、Thread4四条线程分别统计C、D、E、F四个盘的大小，所有线程都统计完毕交给Thread5线程去做汇总，应当如何实现？

这个问题关键就是要知道四条线程何时执行完。

下面是我的解决思路：

/**
 * 如有Thread1、Thread2、Thread3、Thread4四条线程分别统计C、D、E、F四个盘的大小，
 * 所有线程都统计完毕交给Thread5线程去做汇总，应当如何实现？
 *
 * Created by KronChan on 2018/5/14 17:00.
 */
public class CountDownLatchDemo {
    public static void main(String[] args) throws InterruptedException {
        // 初始化计数器，设置总量i，调用一次countDown(）方法后i的值会减1。
        // 在一个线程中如果调用了await()方法，这个线程就会进入到等待的状态，当参数i为0的时候这个线程才继续执行。
        final CountDownLatch countDownLatch = new CountDownLatch(4);
        Runnable thread1 = () -> {
            try {
                TimeUnit.SECONDS.sleep(2);
                System.out.println("统计 C 盘大小");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            // 统计完成计数器 -1
            countDownLatch.countDown();
        };
        Runnable thread2 = () -> {
            try {
                TimeUnit.SECONDS.sleep(2);
                System.out.println("统计 D 盘大小");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            countDownLatch.countDown();
        };
        Runnable thread3 = () -> {
            try {
                TimeUnit.SECONDS.sleep(2);
                System.out.println("统计 E 盘大小");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            countDownLatch.countDown();
        };
        Runnable thread4 = () -> {
            try {
                TimeUnit.SECONDS.sleep(2);
                System.out.println("统计 F 盘大小");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            countDownLatch.countDown();
        };
        ExecutorService pool = Executors.newFixedThreadPool(4);
        pool.execute(thread1);
        pool.execute(thread2);
        pool.execute(thread3);
        pool.execute(thread4);
        // 等待 i 值为 0 ，等待四条线程执行完毕。
        countDownLatch.await();
        System.out.println("统计完成");
        pool.shutdown();
    }
}

同步屏障CyclicBarrier

CyclicBarrier的字面意思是可循环使用（Cyclic）的屏障（Barrier）。它要做的事情是，让一组线程到达一个屏障（也可以叫同步点）时被阻塞，直到最后一个线程到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续运行。

源码分析

属性

private final ReentrantLock lock = new ReentrantLock();
// 线程协作
private final Condition trip = lock.newCondition();
// 必须同时到达barrier的线程个数。
private final int parties;
// parties个线程到达barrier时，会执行的动作，会让到达屏障中的任意一个线程去执行这个动作。
private final Runnable barrierCommand;
// 控制屏障的循环使用，它是可重复使用的，每次使用CyclicBarrier，本次所有线程同属于一代，即同一个Generation
private Generation generation = new Generation();
 // 处在等待状态的线程个数。
private int count;

主要的方法

构造函数

public CyclicBarrier(int parties) {
    this(parties, null);
}
// 构造函数主要实现了，设置一组线程的数量，到达屏障时候的临界点，可以设置到达屏障的时候需要处理的动作，后面屏障允许它们通过。
public CyclicBarrier(int parties, Runnable barrierAction) {
    if (parties <= 0) throw new IllegalArgumentException();
    this.parties = parties;
    this.count = parties;
    this.barrierCommand = barrierAction;
}

await

public int await() throws InterruptedException, BrokenBarrierException {
    try {
        return dowait(false, 0L);
    } catch (TimeoutException toe) {
        throw new Error(toe); // cannot happen;
    }
}
private int dowait(boolean timed, long nanos)
    throws InterruptedException, BrokenBarrierException,
           TimeoutException {
    final ReentrantLock lock = this.lock;
    // 独占锁
    lock.lock();
    try {
        // 保存当前的generation
        final Generation g = generation;
        // generation broken，不允许使用，则抛出异常。
        if (g.broken)
            throw new BrokenBarrierException();
        // 如果当前线程被中断，则通过breakBarrier()终止CyclicBarrier，唤醒CyclicBarrier中所有等待线程。
        if (Thread.interrupted()) {
            breakBarrier();
            throw new InterruptedException();
        }
       // 等待的计数器减一
       int index = --count;
       // 如果计数器的 count 正好为0， 说明已经有parties个线程到达barrier了。执行预定的Runnable任务后，更新换代，准备下一次使用。
       if (index == 0) {  // tripped
           boolean ranAction = false;
           try {
               // 如果barrierCommand不为null，则执行该动作。
               final Runnable command = barrierCommand;
               if (command != null)
                   command.run();
               ranAction = true;
               // 唤醒所有等待线程，并更新generation，准备下一次使用
               nextGeneration();
               return 0;
           } finally {
               if (!ranAction)
                   breakBarrier();
           }
       }
        // 当前线程一直阻塞，
        // 1. 有parties个线程到达barrier
        // 2. 当前线程被中断
        // 3. 超时
        // 直到上面三者之一发生，就唤醒所有线程继续执行下去
        // 自旋
        for (;;) {
            try {
                // 如果不是超时等待，则调用awati()进行等待；否则，调用awaitNanos()进行等待。
                if (!timed)
                    trip.await();
                else if (nanos > 0L)
                    nanos = trip.awaitNanos(nanos);
            } catch (InterruptedException ie) {
                // 如果等待过程中，线程被中断，通过breakBarrier()终止CyclicBarrier，唤醒CyclicBarrier中所有等待线
                if (g == generation && ! g.broken) {
                    breakBarrier();
                    throw ie;
                } else {
                    Thread.currentThread().interrupt();
                }
            }
            // borken
            if (g.broken)
                throw new BrokenBarrierException();
            // 如果generation已经换代，则返回index。
            if (g != generation)
                return index;
            // 超时，则通过breakBarrier()终止CyclicBarrier，唤醒CyclicBarrier中所有等待线程。
            if (timed && nanos <= 0L) {
                breakBarrier();
                throw new TimeoutException();
            }
        }
    } finally {
        lock.unlock(); // 释放独占锁
    }
}
// barrier被broken后，调用breakBarrier方法，将generation.broken设置为true，并使用signalAll通知所有等待的线程。
private void breakBarrier() {
    generation.broken = true;
    count = parties;
    trip.signalAll();
}

使用场景

CyclicBarrier可以用于多线程计算数据，最后合并计算结果的场景，然后四条线程又可以分别去干自己的事情了。

现在我将上面的统计磁盘的任务 CountDownLatch 中改下，统计完统计最终后，每个线程要发出退出信号。

下面是我的实现代码：

public class CyclicBarrierDemo {
    public static void main(String[] args) {
        String[] drivers = {"C", "D", "E", "F"};
        int length = drivers.length;
        ExecutorService pool = Executors.newFixedThreadPool(length);
        // 如果线程都到达barrier状态后，会从四个线程中选择一个线程去执行Runnable。
        CyclicBarrier cyclicBarrier = new CyclicBarrier(length, () -> {
            System.out.printf("%s 线程告诉你，统计完毕，待继续执行%n", Thread.currentThread().getName());
        });
        Stream.of(drivers).forEach((d) -> {
            pool.execute(new StatisticsDemo(d, cyclicBarrier));
        });
        pool.shutdown();
    }
    static class StatisticsDemo implements Runnable {
        private String driveName;
        private CyclicBarrier cyclicBarrier;
        public StatisticsDemo(String driveName, CyclicBarrier cyclicBarrier) {
            this.driveName = driveName;
            this.cyclicBarrier = cyclicBarrier;
        }
        @Override
        public void run() {
            try {
                TimeUnit.SECONDS.sleep((int) (Math.random() * 10));
                System.out.printf("%s 线程统计 %s 盘大小%n", Thread.currentThread().getName(), driveName);
                cyclicBarrier.await();
            } catch (InterruptedException | BrokenBarrierException e) {
                e.printStackTrace();
            }
            System.out.printf("%s 准备退出%n", driveName);
        }
    }
}

执行结果：

pool-1-thread-1 线程统计 C 盘大小
pool-1-thread-2 线程统计 D 盘大小
pool-1-thread-3 线程统计 E 盘大小
pool-1-thread-4 线程统计 F 盘大小
pool-1-thread-4 线程告诉你，统计完毕，待继续执行
F 准备退出
E 准备退出
D 准备退出
C 准备退出

控制并发线程数的Semaphore

Semaphore（信号量）是用来控制同时访问特定资源的线程数量（许可证数），它通过协调各个线程，以保证合理的使用公共资源。

源码分析

构造函数

public Semaphore(int permits) {
     sync = new NonfairSync(permits);
}
public Semaphore(int permits, boolean fair) {
    sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}

具有公平锁的特性，permits 指定许可数量，就是资源数量 state。

同步器的实现

  abstract static class Sync extends AbstractQueuedSynchronizer {
      private static final long serialVersionUID = 1192457210091910933L;
      Sync(int permits) {
          setState(permits);
      }
      final int getPermits() {
          return getState();
      }
// 非公平的方式获取共享锁
      final int nonfairTryAcquireShared(int acquires) {
          for (;;) {
              // 获取资源数量
              int available = getState();
              int remaining = available - acquires; // 本次请求获取锁需要的资源的数量
              if (remaining < 0 ||
                  compareAndSetState(available, remaining)) // 如果资源足够，尝试 CAS 获取锁
                  return remaining;
          }
      }
// 释放锁
      protected final boolean tryReleaseShared(int releases) {
          for (;;) {
              int current = getState();
              int next = current + releases; // 释放锁的时候，返还资源
              if (next < current) // overflow
                  throw new Error("Maximum permit count exceeded");
              if (compareAndSetState(current, next)) // CAS 操作，避免其他的线程也在释放资源
                  return true;
          }
      }
// 减少资源数量
      final void reducePermits(int reductions) {
          for (;;) {
              int current = getState();
              int next = current - reductions;
              if (next > current) // underflow
                  throw new Error("Permit count underflow");
              if (compareAndSetState(current, next))
                  return;
          }
      }
// 清空资源，返回历史资源数量
      final int drainPermits() {
          for (;;) {
              int current = getState();
              if (current == 0 || compareAndSetState(current, 0))
                  return current;
          }
      }
  }
  /**
   * NonFair version
   */
  static final class NonfairSync extends Sync {
      private static final long serialVersionUID = -2694183684443567898L;
      NonfairSync(int permits) {
          super(permits);
      }
      protected int tryAcquireShared(int acquires) {
          return nonfairTryAcquireShared(acquires);
      }
  }
  /**
   * Fair version
   */
  static final class FairSync extends Sync {
      private static final long serialVersionUID = 2014338818796000944L;
      FairSync(int permits) {
          super(permits);
      }
      protected int tryAcquireShared(int acquires) {
          for (;;) {
              // 同样的公平锁情况下，判断该线程前面有没有线程等待获取锁
              if (hasQueuedPredecessors())
                  return -1;
              int available = getState();
              int remaining = available - acquires;
              if (remaining < 0 ||
                  compareAndSetState(available, remaining))
                  return remaining;
          }
      }
  }

提供其他的方法

availablePermits：获取此信号量中当前可用的许可证数（还能有多少个线程执行）；
drainPermits：立刻使用完所有可用的许可证；
reducePermits：减少相应数量的许可证，是一个 protected 方法；
isFair：是否是公平状态；
hasQueuedThreads：等待队列中是否有线程，等待获取许可证；
getQueueLength：等待队列中等待获取许可证的线程数量；
getQueuedThreads：protected 方法，获取等待队列中的线程。

使用场景

Semaphore可以用于做流量控制，特别是公用资源有限的应用场景，比如我们有五台机器，有十名工人，每个工人需要一台机器才能工作，一名工人工作完了就可以休息了，机器让其他没工作过的工人使用。

下面是我的实现代码：

public class SemaphoreDemo {
    public static void main(String[] args) {
        int num = 10;
        Semaphore machines = new Semaphore(5);
        for (int i = 0; i < num; i++) {
            new Thread(new Worker(i, machines)).start();
        }
    }
    static class Worker extends Thread {
        private Semaphore machines;
        private int worker;
        Worker(int worker, Semaphore semaphore) {
            this.worker = worker;
            this.machines = semaphore;
        }
        @Override
        public void run() {
            try {
                machines.acquire();
                System.out.printf("工人 %d 开始使用机器工作了 %n", worker);
                TimeUnit.SECONDS.sleep((int) (Math.random() * 10));
                System.out.printf("工人 %d 干完活了，让出机器了%n", worker);
                machines.release();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

执行一下结果：

工人 0 开始使用机器工作了 
工人 4 开始使用机器工作了 
工人 3 开始使用机器工作了 
工人 2 开始使用机器工作了 
工人 1 开始使用机器工作了 
工人 1 干完活了，让出机器了
工人 5 开始使用机器工作了 
工人 5 干完活了，让出机器了
工人 6 开始使用机器工作了 
工人 2 干完活了，让出机器了
工人 7 开始使用机器工作了 
工人 4 干完活了，让出机器了
工人 8 开始使用机器工作了 
工人 0 干完活了，让出机器了
工人 9 开始使用机器工作了 
工人 8 干完活了，让出机器了
工人 6 干完活了，让出机器了
工人 3 干完活了，让出机器了
工人 9 干完活了，让出机器了
工人 7 干完活了，让出机器了

虽然上面有 10 个工人（线程）一起并发，但是，它同时只有五个工人能够是执行的。

线程间交换数据的Exchanger

Exchanger（交换者）是一个用于线程间协作的工具类。Exchanger 用于两个工作线程间的数据交换。

具体上来说，Exchanger类允许在两个线程之间定义同步点。当两个线程都到达同步点时，他们交换数据结构，因此第一个线程的数据进入到第二个线程中，第二个线程的数据进入到第一个线程中，这要就完成了一个“交易”的环节。

源码分析

源码很难看懂，主要还是

【死磕Java并发】—–J.U.C之并发工具类：Exchanger

使用场景

Exchanger 可以用于遗传算法。遗传算法里需要选出两个人作为交配对象，这时候会交换两人的数据。

下面做一个卖书买书的例子：

public class ExchangerDemo {
    private static final Exchanger<String> EXCHANGER = new Exchanger<>();
    private static final ExecutorService POOLS = Executors.newFixedThreadPool(2);
    public static void main(String[] args) throws InterruptedException {
        POOLS.execute(() -> {
            String bookName = "浮生六记";
            System.out.printf("饭饭要卖一本%s。%n", bookName);
            try {
                String pay = EXCHANGER.exchange(bookName);
                System.out.printf("饭饭卖出一本%s赚了%s￥。%n", bookName, pay);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        TimeUnit.SECONDS.sleep(5);
        System.out.println("》》》》》》》》饭饭先到了交易地点 睡了 5 s，七巧来了");
        System.out.println("》》》》》》》》准备交易");
        POOLS.execute(() -> {
            String pay = "23";
            try {
                String bookName = EXCHANGER.exchange(pay);
                System.out.printf("七巧付了%s￥买了一本%s。%n", pay, bookName);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        POOLS.shutdown();
        for (; ; ) {
            if (POOLS.isTerminated()) {
                System.out.println("交易结束！");
                return;
            }
        }
    }
}

执行结果：

饭饭要卖一本浮生六记。
》》》》》》》》饭饭先到了交易地点 睡了 5 s，七巧来了
》》》》》》》》准备交易
七巧付了23￥买了一本浮生六记。
饭饭卖出一本浮生六记赚了23￥。
交易结束！

总结

Exchanger主要完成的是两个工作线程之间的数据交换，如果有一个线程没有执行 exchange()方法，则会一直等待。还可以设置最大等待时间exchange（V v, TimeUnit unit）

CyclicBarrier和CountDownLatch的区别

CountDownLatch的计数器只能使用一次，而CyclicBarrier的计数器可以使用reset()方法重置。所以CyclicBarrier能处理更为复杂的业务场景。例如，如果计算发生错误，可以重置计数器，并让线程重新执行一次。
CyclicBarrier还提供其他有用的方法，比如getNumberWaiting方法可以获得 CyclicBarrier阻塞的线程数量。isBroken()方法用来了解阻塞的线程是否被中断。

参考文章

《Java 并发编程的艺术》

------本文结束感谢阅读------

 Java 并发编程同步器 AQS

创建时间：	2018/10/29 9:31
来源：	https://blog.wuwii.com/juc-utils.html